package com.example.repository;

import com.example.model.Review;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.stereotype.Repository;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

@Repository
public class ReviewRepository {

    private final Connection hbaseConnection;
    private final String tableName = "reviews";
    private final byte[] reviewInfoFamily = Bytes.toBytes("review_info");
    private final byte[] userInfoFamily = Bytes.toBytes("user_info");
    private final byte[] productInfoFamily = Bytes.toBytes("product_info");


    public ReviewRepository(Connection hbaseConnection) {
        this.hbaseConnection = hbaseConnection;
    }

    public void save(Review review) throws IOException {
        try (Table table = hbaseConnection.getTable(TableName.valueOf(tableName))) {
            long reverseTimestamp = Long.MAX_VALUE - System.currentTimeMillis();
            String rowKey = review.getProductId() + "_" + reverseTimestamp;
            review.setRowKey(rowKey); // Set the generated rowkey back to the object

            Put put = new Put(Bytes.toBytes(rowKey));

            // Add review info
            put.addColumn(reviewInfoFamily, Bytes.toBytes("userId"), Bytes.toBytes(review.getUserId()));
            put.addColumn(reviewInfoFamily, Bytes.toBytes("rating"), Bytes.toBytes(String.valueOf(review.getRating())));
            put.addColumn(reviewInfoFamily, Bytes.toBytes("comment"), Bytes.toBytes(review.getComment()));
            put.addColumn(reviewInfoFamily, Bytes.toBytes("reviewDate"), Bytes.toBytes(String.valueOf(System.currentTimeMillis())));

            // Add user info
            if (review.getUserName() != null) {
                put.addColumn(userInfoFamily, Bytes.toBytes("userName"), Bytes.toBytes(review.getUserName()));
            }

            // Add product info
            if (review.getProductName() != null) {
                put.addColumn(productInfoFamily, Bytes.toBytes("productName"), Bytes.toBytes(review.getProductName()));
            }
            if (review.getProductCategory() != null) {
                put.addColumn(productInfoFamily, Bytes.toBytes("productCategory"), Bytes.toBytes(review.getProductCategory()));
            }

            table.put(put);
        }
    }

    public List<Review> findByProductId(String productId) throws IOException {
        List<Review> reviews = new ArrayList<>();
        try (Table table = hbaseConnection.getTable(TableName.valueOf(tableName))) {
            Scan scan = new Scan();
            scan.withStartRow(Bytes.toBytes(productId + "_"));
            scan.withStopRow(Bytes.toBytes(productId + "~"));

            for (Result result : table.getScanner(scan)) {
                if (result.isEmpty()) continue;

                Review review = new Review();
                review.setRowKey(Bytes.toString(result.getRow()));
                review.setProductId(productId);

                // review_info family - defensive reading
                byte[] userIdBytes = result.getValue(reviewInfoFamily, Bytes.toBytes("userId"));
                if (userIdBytes != null) review.setUserId(Bytes.toString(userIdBytes));

                byte[] ratingBytes = result.getValue(reviewInfoFamily, Bytes.toBytes("rating"));
                if (ratingBytes != null) review.setRating(Integer.parseInt(Bytes.toString(ratingBytes)));

                byte[] commentBytes = result.getValue(reviewInfoFamily, Bytes.toBytes("comment"));
                if (commentBytes != null) review.setComment(Bytes.toString(commentBytes));
                
                byte[] reviewDateBytes = result.getValue(reviewInfoFamily, Bytes.toBytes("reviewDate"));
                if (reviewDateBytes != null) review.setReviewDate(Bytes.toString(reviewDateBytes));

                // user_info family - defensive reading
                byte[] userNameBytes = result.getValue(userInfoFamily, Bytes.toBytes("userName"));
                if (userNameBytes != null) review.setUserName(Bytes.toString(userNameBytes));

                // product_info family
                byte[] productNameBytes = result.getValue(productInfoFamily, Bytes.toBytes("productName"));
                if (productNameBytes != null) review.setProductName(Bytes.toString(productNameBytes));
                
                byte[] productCategoryBytes = result.getValue(productInfoFamily, Bytes.toBytes("productCategory"));
                if (productCategoryBytes != null) review.setProductCategory(Bytes.toString(productCategoryBytes));
                
                reviews.add(review);
            }
        }
        return reviews;
    }

    // Add methods for find, delete etc.
} 